package chan.module.project.hw.mrs.hdfs.v331;

import chan.module.project.comm.PropertiesUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;

public class MRSHdfsV331Main {
    private static final Logger LOG = LogManager.getLogger(MRSHdfsV331Main.class);

    private FileSystem fSystem;


    public static String getUserRealm() {
        String serverRealm = System.getProperty("SERVER_REALM");
        String AUTH_HOST_NAME;
        if (serverRealm != null && serverRealm != "") {
            AUTH_HOST_NAME = "hadoop." + serverRealm.toLowerCase();
        } else {
            serverRealm = KerberosUtil.getKrb5DomainRealm();
            if (serverRealm != null && serverRealm != "") {
                AUTH_HOST_NAME = "hadoop." + serverRealm.toLowerCase();
            } else {
                AUTH_HOST_NAME = "hadoop";
            }
        }
        return AUTH_HOST_NAME;
    }


    public static void main(String[] args) throws Exception {
        PropertiesUtil util = PropertiesUtil.readConfig(args);
        // 完成初始化和认证
        String PATH_TO_KRB5_CONF = util.get("krb5.conf");
        System.setProperty("java.security.krb5.conf", PATH_TO_KRB5_CONF);
        System.setProperty("sun.security.krb5.debug", "true");
        String encryption_types = util.get("sun.security.krb5.encryption.types");
        System.setProperty("sun.security.krb5.encryption.types",encryption_types);


        String user_principal = util.get("user.prncipal");
        String path_to_keytab = util.get("user.keytab");
        String krb5_cdc = util.get("krb5.cdc");
        String krb5_realm = util.get("krb5.realm");
        String path_to_hdfs_site_xml = util.get("hdfs.site.xml");
        String path_to_core_site_xml = util.get("core.site.xml");
        Configuration configuration = new Configuration();
        // conf file
        configuration.addResource(new Path(path_to_hdfs_site_xml));
        configuration.addResource(new Path(path_to_core_site_xml));
        configuration.set("hadoop.security.authentication", "kerberos");
        configuration.set("hadoop.security.authorization", "true");
        configuration.set("hadoop.security.kerberos.krb5.kdc",krb5_cdc); // KDC 地址
        configuration.set("hadoop.security.kerberos.krb5.realm",krb5_realm); // Realm 名称
        configuration.set("user.kerberos.keytab", path_to_keytab); // keytab 路径
        configuration.set("user.kerberos.principal",user_principal); // 主体名

        // 显式指定 Kerberos 加密类型（关键修复点）
        configuration.set("hadoop.security.kerberos.krb5.kdc", krb5_cdc); // KDC 地址
        configuration.set("hadoop.security.kerberos.krb5.realm",krb5_realm); // Realm 名称

        LOG.info("开始登录HDFS......");
        UserGroupInformation.setConfiguration(configuration);
        UserGroupInformation.loginUserFromKeytab(user_principal,path_to_keytab);

        // 安全模式 登录认证
        //LoginUtil.login(USER_PRINCIPAL,PATH_TO_KEYTAB,PATH_TO_KRB5_CONF,configuration);
        UserGroupInformation loginUser = UserGroupInformation.getLoginUser();

        LOG.info("是否已经登录hadoop：{}",loginUser.isFromKeytab() ? "是" : "否");
        LOG.info("当前用户为：{}",loginUser);

        // 业务示例1：一个普通用例
        String DEFAULT_FS = util.get("hdfs.defaultFS");
        MRSHdfsV331Main hdfsExample = new MRSHdfsV331Main();
        LOG.info("获取HDFS对象文件系统。");
        hdfsExample.fSystem = FileSystem.get(URI.create(DEFAULT_FS),configuration,user_principal);

        String hdfsTestDir = util.get("hdfs.test.dir");
        String hdfsTestFileName = util.get("hdfs.test.fileName");
        LOG.info("创建目录:"+hdfsTestDir);
        hdfsExample.createPath(hdfsTestDir);
        String fileName = hdfsTestDir+ File.separator + hdfsTestFileName;
        hdfsExample.deletePath(fileName);
        LOG.info("创建文件:"+fileName);
        hdfsExample.write(fileName);
        LOG.info("读取文件:"+fileName);
        hdfsExample.read(fileName);
    }

    private void write(String fileName) throws Exception {
        final String content = "hi, I am bigdata. It is successful if you can see me:" + System.currentTimeMillis();
        FSDataOutputStream out = null;
        try {
            out = fSystem.create(new Path(fileName));
            out.write(content.getBytes());
            out.hsync();
            LOG.info("success to write.");
        } finally {
            // make sure the stream is closed finally.
            IOUtils.closeStream(out);
        }
    }


    /**
     * read file
     *
     * @throws IOException
     */
    private void read(String fileName) throws IOException {
        Path path = new Path(fileName);
        FSDataInputStream in = null;
        BufferedReader reader = null;
        StringBuffer strBuffer = new StringBuffer();
        try {
            in = fSystem.open(path);
            reader = new BufferedReader(new InputStreamReader(in));
            String sTempOneLine;
            // write file
            while ((sTempOneLine = reader.readLine()) != null) {
                strBuffer.append(sTempOneLine);
            }
            LOG.info("result is : " + strBuffer.toString());
            LOG.info("success to read.");
        } finally {
            // make sure the streams are closed finally.
            IOUtils.closeStream(reader);
            IOUtils.closeStream(in);
        }
    }

    /**
     * create file path
     * @return
     * @throws IOException
     */
    private boolean createPath(String dir) throws IOException {
        final Path filePath = new Path(dir);
        if (!fSystem.exists(filePath)) {
            fSystem.mkdirs(filePath);
        }
        return true;
    }

    /**
     * delete file path
     * @return
     * @throws IOException
     */
    private boolean deletePath(String fileName) throws IOException {
        final Path filePath = new Path(fileName);
        if (!fSystem.exists(filePath)) {
            return false;
        }
        return fSystem.delete(filePath, true);
    }

}
